package com.xia.im.netty.wschat;

import cn.hutool.core.date.DateUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ChatServer {
    public static void main(String[] args) {
        start();
    }
    private static void start() {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 创建子线程组，专门负责IO任务的处理
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workGroup);
            bootstrap.channel(NioServerSocketChannel.class);
            //继承自ChannelInitializer
            bootstrap.childHandler(new WebSocketChannelInitializer());
            log.info("Chat服务已启动....");
            Channel ch = bootstrap.bind(9090).sync().channel();
            ch.closeFuture().sync();


        } catch (Exception e) {
            e.printStackTrace();
        } finally {
//            //退出程序
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

/**
 * 总体来说
 * WebSocket 的初始握手协议是 HTTP 协议
 * 握手完成后，Netty 会自动移除 HTTP 处理器，仅保留 WebSocket 处理器
 *
 *  HttpServerCodec         握手阶段 后续移除
 *  HttpObjectAggregator    握手阶段 后续移除
 *  ChunkedWriteHandler     大数据传输   一直存在，处理大文件用
 */
class WebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        //http请求编码器,包含两个组成部分
        //HttpRequestDecoder,将字节流解码为 HttpRequest + HttpContent
        //HttpResponseEncoder,将 HttpResponse + HttpContent 编码为字节流
        //为什么需要它,WebSocket 协议通过 HTTP 协议升级（Upgrade）而来，初始握手是 HTTP 请求
        ch.pipeline().addLast("http-codec",new HttpServerCodec());//设置解码器
        //HTTP 消息聚合器,将多个 HttpContent 碎片合并为完整的 FullHttpRequest 或 FullHttpResponse
        //参数 65536：指定最大聚合内容长度（字节），超过会抛异常
        //为什么需要它,HTTP 请求可能分多次传输（如大文件上传）,WebSocket 握手需要完整的 HTTP 头部和内容（如 Sec-WebSocket-Key）
        ch.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));//聚合器，使用websocket会用到
        //支持分块传输（如大文件、流式数据）
        //为什么需要它,WebSocket 可能传输大数据（如视频分片）,防止大文件占用过多内存（按块写入网络）
        ch.pipeline().addLast("http-chunked",new ChunkedWriteHandler());//用于大数据的分区传输
        ch.pipeline().addLast(new ChatEngine());
    }
}

@Slf4j
class ChatEngine extends SimpleChannelInboundHandler<Object> {
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private WebSocketServerHandshaker handshaker;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 处理http和websocket消息
        if (msg instanceof FullHttpRequest){
            //以http请求形式接入，但是走的是websocket
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        }else if (msg instanceof WebSocketFrame){
            //处理websocket客户端的消息
            handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 唯一的一次http请求，用于创建websocket
     * */
    private void handleHttpRequest(ChannelHandlerContext ctx,
                                   FullHttpRequest req) {
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws://localhost:9090/websocket", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory
                    .sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame){
        // 判断是否关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(
                    new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 返回应答消息
        String msg = ((TextWebSocketFrame) frame).text();

        channelGroup.forEach(channel -> {
            if (channel!=ctx.channel()){
                TextWebSocketFrame tws = new TextWebSocketFrame("客户端【"+ctx.channel().id()+"】"+ DateUtil.now()
                        +"：" + msg);
                channel.writeAndFlush(tws);
            }else{
                TextWebSocketFrame tws = new TextWebSocketFrame("我【"+ctx.channel().id()+"】"+DateUtil.now()
                        +"：" + msg);
                channel.writeAndFlush(tws);
            }
        });
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String message = String.format("大家好，有新的成员(%s)加入了,欢迎欢迎", ctx.channel().remoteAddress());
        if (!channelGroup.isEmpty()){
            TextWebSocketFrame tws = new TextWebSocketFrame("系统【"+ctx.channel().id()+"】"+DateUtil.now()
                    +"：" + message);
            channelGroup.writeAndFlush(tws);
        }
        log.info(message);
        channelGroup.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        String message = String.format("大家好，有成员(%s)下线了...%s", ctx.channel().remoteAddress(), DateUtil.now());
        if (!channelGroup.isEmpty()){
            TextWebSocketFrame tws = new TextWebSocketFrame(DateUtil.now()
                    + ctx.channel().id() + "：" + message);
            channelGroup.writeAndFlush(tws);
        }
        log.info(message);
    }
}
